package com.we.flink.userlabelplatform;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;

import com.alibaba.fastjson.JSONObject;
import com.we.flink.userlabelplatform.table.*;
import com.we.flink.userlabelplatform.tag.*;
import com.we.flink.utils.WeKafkaPropertyReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class DimUserLabel {
    public static final String RELEASEPROP =
            "userlabelplatform/prod_adm_business_user_label.properties";
    public static final String KEY_BY = "uid";
    public static Logger LOG = LoggerFactory.getLogger(DimUserLabel.class);

    public static void main(String[] args) {
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        if (parameterTool.getNumberOfParameters() == 0) {
            supportedTables();
            return;
        }

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment envTable = StreamTableEnvironment.create(env);
        WeKafkaPropertyReader propReader = WeKafkaPropertyReader.init(RELEASEPROP);
        try {
            String kfkBootStrapServer = propReader.getTmpKfkBootStrapServer();
            String kfkTopic = propReader.getTmpKfkTopic();
            int kfkPartitions = propReader.getTmpKfkPartitions();

            env.setStateBackend(new RocksDBStateBackend(propReader.getRocksDBBackendUrl()));

            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
            ckConf.enableUnalignedCheckpoints();

            /** sink kafka property */
            Properties sinkProp = new Properties();
            sinkProp.setProperty("bootstrap.servers", kfkBootStrapServer);
            sinkProp.setProperty("acks", "all");
            /** hive catlog configure */
            String hiveCatlogName = propReader.getHiveCatlogName();
            String hiveDBName = propReader.getHiveDBName();
            String hiveConfDir = propReader.getHiveConfDir();

            HiveCatalog hiveCatalog = new HiveCatalog(hiveCatlogName, hiveDBName, hiveConfDir);
            envTable.registerCatalog("hive", hiveCatalog);
            envTable.useCatalog("hive");

            /** sink to Kafka */
            FlinkKafkaProducer<String> userlabelKafkaProducer =
                    new FlinkKafkaProducer<String>(
                            kfkTopic,
                            new WeKafkaKeyedSerializationSchema(),
                            sinkProp,
                            java.util.Optional.of(new WeKafkaCustomPartitioner()));
            if (parameterTool.has("hfqaudithomepagelog")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> HfqAuditHomePageLogflatMapStream =
                        TableHfqAuditHomePageLog.getFlatMapStream(envTable, env);
                HfqAuditHomePageLogflatMapStream.addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            // 7kw
            if (parameterTool.has("lendauditrecord")
                    || parameterTool.has("risklabel")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> lendAuditRecordFlatMapStream =
                        TagLendAuditRecord.getFlatMapStream(envTable, env);
                lendAuditRecordFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            if (parameterTool.has("offlinelabel")
                    || parameterTool.has("risklabel")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> offlineLabelFlatMapStream =
                        TableOfflineLabel.getFlatMapStream(envTable, env);
                offlineLabelFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }

            if (parameterTool.has("orderinfo")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> orderInfoFlatMapStream =
                        TagOrderInfo.getFlatMapStream(envTable, env);
                orderInfoFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            // 10kw
            if (parameterTool.has("risktag")
                    || parameterTool.has("part2")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> riskTagFlatMapStream =
                        TagRiskTag.getFlatMapStream(envTable, env);
                riskTagFlatMapStream.addSink(userlabelKafkaProducer).setParallelism(kfkPartitions);
            }
            if (parameterTool.has("savemoneycardmember")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> saveMonCardMemFlatMapStream =
                        TagSaveMoneyCardMember.getFlatMapStream(envTable, env);
                saveMonCardMemFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            if (parameterTool.has("savemoncardmemspecial")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> saveMonCardMemSpecialPriceFlatMapStream =
                        TagSaveMoneyCardMemberSpecialPrice.getFlatMapStream(envTable, env);
                saveMonCardMemSpecialPriceFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            if (parameterTool.has("supermemrecord")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> superMemRecordFlatMapStream =
                        TagSuperMemberRecord.getFlatMapStream(envTable, env);
                superMemRecordFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            if (parameterTool.has("tiecardamtrcd")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> tieCardAmtRcdFlatMapStream =
                        TagTieCardAmountRecord.getFlatMapStream(envTable, env);
                tieCardAmtRcdFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            if (parameterTool.has("tiecardrisktag")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> tieCardRiskTag =
                        TagTieCardRiskTag.getFlatMapStream(envTable, env);
                tieCardRiskTag.addSink(userlabelKafkaProducer).setParallelism(kfkPartitions);
            }
            if (parameterTool.has("txxyorderrescallback")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> txxyOrderResCallbackFlatMapStream =
                        TagTxxyOrderResultCallback.getFlatMapStream(envTable, env);
                txxyOrderResCallbackFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            if (parameterTool.has("useraccount") || parameterTool.has("risklabel")) {
                SingleOutputStreamOperator<String> userAccountflatMapStream =
                        TagUserAccount.getFlatMapStream(envTable, env);
                userAccountflatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            // 3kw
            if (parameterTool.has("useraddress")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> userAddressflatMapStream =
                        TableUserAddress.getFlatMapStream(envTable, env);
                userAddressflatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            if (parameterTool.has("useramtretrialadt")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> userAmtRetrialadtflatMapStream =
                        TagUserAmountRetrialAudit.getFlatMapStream(envTable, env);
                userAmtRetrialadtflatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            // 3kw
            if (parameterTool.has("useraudit")
                    || parameterTool.has("risklabel")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> userAuditFlatMapStream =
                        TagUserAudit.getFlatMapStream(envTable, env);
                userAuditFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            if (parameterTool.has("usercertifycmpy")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> userCertifyCompanyFlatMapStream =
                        TableUserCertifyCompany.getFlatMapStream(envTable, env);
                userCertifyCompanyFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            if (parameterTool.has("usercertifyedu")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> userCertifyEduFlatMapStream =
                        TableUserCertifyEducation.getFlatMapStream(envTable, env);
                userCertifyEduFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            if (parameterTool.has("userentrepreneur")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> userEntrepreneurRecordFlatMapStream =
                        TableUserEntrepreneurRecord.getFlatMapStream(envTable, env);
                userEntrepreneurRecordFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            // 2kw
            if (parameterTool.has("userlevel")
                    || parameterTool.has("part2")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> userLevelFlatMapStream =
                        TagUserLevel.getFlatMapStream(envTable, env);
                userLevelFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }
            // 8kw
            if (parameterTool.has("userloan")
                    || parameterTool.has("risklabel")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> userLoanFlatMapStream =
                        TagUserLoan.getFlatMapStream(envTable, env);
                userLoanFlatMapStream.addSink(userlabelKafkaProducer).setParallelism(kfkPartitions);
            }

            if (parameterTool.has("wechathfquserinfo")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> wechatHfqUserInfoFlatMapStream =
                        TagWechatHaofenqiUserInfo.getFlatMapStream(envTable, env);
                wechatHfqUserInfoFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }

            if (parameterTool.has("biguocardrisktag")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> biguoCardRiskTagFlatMapStream =
                        TagBiguoCardRiskTag.getFlatMapStream(envTable, env);
                biguoCardRiskTagFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }

            if (parameterTool.has("jiangfeicardrisktag")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> jiangfeiCardRiskTagFlatMapStream =
                        TagJiangfeiCardRiskTag.getFlatMapStream(envTable, env);
                jiangfeiCardRiskTagFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }

            if (parameterTool.has("quotachangerecord")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> quotaChangeRecordFlatMapStream =
                        TagQuotaChangeRecord.getFlatMapStream(envTable, env);
                quotaChangeRecordFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }

            if (parameterTool.has("temptofixedrisktag")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> tempToFixedRiskTagFlatMapStream =
                        TagTempToFixedRiskTag.getFlatMapStream(envTable, env);
                tempToFixedRiskTagFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }

            if (parameterTool.has("loansubmitrepayrecordtag")
                    || parameterTool.has("part2")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> loanSubmitRepayRecordTagFlatMapStream =
                        TagLoanSubmitRepayRecord.getFlatMapStream(envTable, env);
                loanSubmitRepayRecordTagFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }

            if (parameterTool.has("hfqUserstrategytag")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> hfqUserStrategyTagFlatMapStream =
                        TagHfqUserStrategy.getFlatMapStream(envTable, env);
                hfqUserStrategyTagFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }

            if (parameterTool.has("usermembertag")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> userMemberTagFlatMapStream =
                        TagUserMember.getFlatMapStream(envTable, env);
                userMemberTagFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }

            if (parameterTool.has("incrlimitrecordtag")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> incrLimitRecordFlatMapStream =
                        TagIncrLimitRecord.getFlatMapStream(envTable, env);
                incrLimitRecordFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }

            if (parameterTool.has("userphoneinfotag")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> userPhoneInfoFlatMapStream =
                        TagUserPhoneInfo.getFlatMapStream(envTable, env);
                userPhoneInfoFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }

            if (parameterTool.has("dccarloantag")
                    || parameterTool.has("part1")
                    || parameterTool.has("all")) {
                SingleOutputStreamOperator<String> dcCarLoanFlatMapStream =
                        TagDcCarLoan.getFlatMapStream(envTable, env);
                dcCarLoanFlatMapStream
                        .addSink(userlabelKafkaProducer)
                        .setParallelism(kfkPartitions);
            }

            /** submit the application */
            env.execute(DimUserLabel.class.toString());
        } catch (Exception E) {
            LOG.error("Exception: " + E.getMessage());
            E.printStackTrace();
        }
    }

    public static void supportedTables() {
        System.out.println("no table params setted!! ");
        System.out.println(
                "-hfqaudithomepagelog \n"
                        + "-lendauditrecord \n"
                        + "-offlinelabel \n"
                        + "-orderinfo \n"
                        + "-risktag \n"
                        + "-savemoneycardmember \n"
                        + "-savemoncardmemspecial \n"
                        + "-supermemrecord \n"
                        + "-txxyorderrescallback \n"
                        + "-useraccount \n"
                        + "-useraddress \n"
                        + "-useraudit \n"
                        + "-usercertifycmpy \n"
                        + "-usercertifyedu \n"
                        + "-userentrepreneur \n"
                        + "-userlevel \n"
                        + "-userloan \n"
                        + "-g1_userlabel \n"
                        + "-g2_lendadrcd_rsktg \n"
                        + "-g3user_act_adrs_adt \n"
                        + "-g4_usrlv_loan \n");
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            Long uid = jsonObject.getLong(KEY_BY);
            return String.valueOf(uid).getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            //            LOG.error("WeKafkaCustomPartitioner partitions: " + partitions.length + "
            // partition: " + partition + " key: " + new String(key));
            return partition;
        }
    }
}
